package rabbitmq

import (
	"fmt"
	"github.com/streadway/amqp"
	"log"
)

const (
	RMQADDR     = "amqp://admin:admin@8.129.4.17:5672/"
	QUEUENAME   = "msgQueueWithPersist"
	PRODUCERCNT = 50
	CONSUMERCNT = 5
)

var conn *amqp.Connection
var channel *amqp.Channel
var queue amqp.Queue

func Connect() (conn *amqp.Connection, err error) {
	conn, err = amqp.Dial(RMQADDR)
	return conn, nil
}

func MakeChannel() (channel *amqp.Channel, err error) {
	conn, err := Connect()
	if err != nil {
		failOnError(err, "Failed to connect to RabbitMQ")
		return
	}
	channel, err = conn.Channel()
	return
}

func failOnError(err error, msg string) {
	if err != nil {
		fmt.Printf("%s: %s\n", msg, err)
	}
}

func MakeQueue(durable bool) (queue amqp.Queue, err error) {
	channel, err := MakeChannel()
	fmt.Println("队列：", err)
	queue, err = channel.QueueDeclare(QUEUENAME, durable, false, false, false, nil)
	return
}

func PublishMessage(message string) {
	conn, err := Connect()
	if err != nil {
		fmt.Println("连接Rabbitmq失败")
		return
	}
	defer conn.Close()
	//go func() {
	channel, err := MakeChannel()
	if err != nil {
		failOnError(err, "创建通道失败")
		return
	}
	defer channel.Close()
	queue, err := MakeQueue(true)
	if err != nil {
		failOnError(err, "创建队列失败")
		return
	}
	fmt.Println("队列名称：", queue.Name)
	err = channel.Publish(
		QUEUENAME,  //exchange
		queue.Name, //routing key
		false,
		false,
		amqp.Publishing{
			DeliveryMode: amqp.Persistent, //Msg set as persistent
			ContentType:  "text/plain",
			Body:         []byte(message),
		})
	log.Printf(" [x] Sent %s", message)
	failOnError(err, "Failed to publish a message")
	if err == nil {
		log.Printf(" [x] Sent %s 成功", message)
	}
	//}()
}

func Consume() {
	conn, err := Connect()
	if err != nil {
		fmt.Println("连接Rabbitmq失败")
		return
	}
	defer conn.Close()

	forever := make(chan bool)
	go func() {
		channel, err := MakeChannel()
		if err != nil {
			fmt.Println("创建Rabbitmq channel失败")
			return
		}
		defer channel.Close()
		queue, err := MakeQueue(true)
		if err != nil {
			fmt.Println("创建Rabbitmq 队列失败,%s", err)
			return
		}
		err = channel.Qos(1, 0, false)
		if err != nil {
			fmt.Println("创建Rabbitmq QOS 失败")
			return
		}
		msgs, err := channel.Consume(
			queue.Name,
			"MsgWorkConsumer",
			false, //Auto Ack
			false,
			false,
			false,
			nil,
		)
		if err != nil {
			log.Fatal(err)
		}

		for msg := range msgs {
			log.Printf("In consume a message: %s\n", msg.Body)
			log.Printf("Done")
			msg.Ack(false) //Ack
		}
	}()
	<-forever
}
